package mxx.mr.job;

import org.apache.commons.codec.digest.DigestUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.MapWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Set;

/**
 * @program:report-form-hadoop
 * @description:设备月平均OEE-Job
 * @author:AlTuring
 * @Date:2019/11/12-16:17
 **/
public class DeviceMonthAverageOeeJob {
    /**
     *设备月平均OEE -Mapper
     * 1.求设备总数
     */
    public static class DeviceMonthAverageOeeTotalMapper extends Mapper<LongWritable,Text,Text, MapWritable> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            MapWritable mw = new MapWritable();
            //因为只有一条数据，所以直接使用固定的key
            Text outKey = new Text("deviceOEE");
            mw.put(new IntWritable(1),value);
            context.write(outKey,mw);
        }
    }

    /**
     * 设备月平均OEE -Mapper
     * 2.设备完好数量
     */
    public static class DeviceMonthAverageOeeIntactTotalMapper extends Mapper<LongWritable,Text,Text,MapWritable>{
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            MapWritable mw = new MapWritable();
            //因为只有一条数据，所以直接使用固定的key
            Text outKey = new Text("deviceOEE");
            mw.put(new IntWritable(2),value);
            context.write(outKey,mw);
        }
    }

    /**
     * 设备月平均OEE -Mapper
     * 3.设备运行数量
     */
    public static class DeviceMonthAverageOeeRunTotalMapper extends Mapper<LongWritable,Text,Text,MapWritable>{
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String[] datas = value.toString().split(",");
            //设备总数
            String deviceSEQ = datas[0];
            String time = datas[1];
            //获取当前的（年-月）:2019-11，作为outKey
            MapWritable mw = new MapWritable();
            //因为只有一条数据，所以直接使用固定的key
            Text outKey = new Text("deviceOEE");
            mw.put(new IntWritable(3),value);

            context.write(outKey,mw);
        }
    }
    /**
     * 设备报表0 -Reducer
     * 设备月平均OEE
     */
    public static class DeviceMonthAverageOeeReducer extends Reducer<Text, MapWritable, Text, Text> {
        @Override
        protected void reduce(Text key, Iterable<MapWritable> values, Context context) throws IOException, InterruptedException {
            int total = 0;
            int good = 0;
            int run = 0;
            //设置时间默认为当前时间（2019-11）
            Date date = new Date();
            SimpleDateFormat dateFormat=new SimpleDateFormat("YYYY-MM");//设置当前时间的格式，为年-月-日
            String time = dateFormat.format(date);
            //定义set存储设备的id，设备的id的数量=设备开动数量
            Set<String> set = new HashSet<>();
            for (MapWritable mw : values){
                if (mw.containsKey(new IntWritable(1))){
                    total = Integer.valueOf(mw.get(new IntWritable(1)).toString());
                }else if (mw.containsKey(new IntWritable(2))){
                    //故障数量
                    good = Integer.valueOf(mw.get(new IntWritable(2)).toString());
                }else if (mw.containsKey(new IntWritable(3))){
                    String[] datas = mw.get(new IntWritable(3)).toString().split(",");
                    set.add(datas[0]);
                    time = datas[1].substring(0,7);
                }
            }
            //设备完好数量 = 设备总数 - 设备故障数量
            good = total -good;
            //设备开动数量 ，只要在设备状态表中有数据，就代表设备开动过。
            run = set.size();
            //自定义outValue（设备总数,设备完好数量,设备运行数量,时间:2019-11）
            String outValue = total +","+
                    good+","+
                    run+","+
                    "chilun"+","+
                    time;

            Text outKey = new Text(DigestUtils.shaHex(key.toString()));

            context.write(outKey, new Text(outValue));
        }
    }
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration configuration = new Configuration();
        configuration.set("mapred.textoutputformat.separator", ",");

        // 如果输出目录已经存在，则先删除
        FileSystem fileSystem = FileSystem.get(configuration);

        // 输出文件
        Path outputPath = new Path(args[0]);
        if (fileSystem.exists(outputPath)) {
            fileSystem.delete(outputPath, true);
        }
        // 输入文件
        Path inputPath1 = new Path(args[1]);
        Path inputPath2 = new Path(args[2]);
        Path inputPath3 = new Path(args[3]);

        // job
        Job job = Job.getInstance(configuration);
        job.setJobName("DeviceMonthAverageOeeJob");
        job.setJarByClass(DeviceMonthAverageOeeJob.class);


        // 多个输入对应多个mapper
        //设备总数
        MultipleInputs.addInputPath(job,inputPath1, TextInputFormat.class, DeviceMonthAverageOeeTotalMapper.class);
        //设备完好数量
        MultipleInputs.addInputPath(job,inputPath2, TextInputFormat.class, DeviceMonthAverageOeeIntactTotalMapper.class);
        //设备运行数量
        MultipleInputs.addInputPath(job,inputPath3, TextInputFormat.class, DeviceMonthAverageOeeRunTotalMapper.class);

        //设置map输出类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(MapWritable.class);
        // 设置reducer输出类型
        job.setReducerClass(DeviceMonthAverageOeeReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(MapWritable.class);
        // 一个输出
        FileOutputFormat.setOutputPath(job,outputPath);
        // 完成
        job.waitForCompletion(true);
        System.out.println("---设备月平均OEE报表Job完成---");
    }
}
